package com.lg.distributed.transaction.message.listener;

import com.lg.distributed.transaction.message.constant.MQConstants;
import com.lg.distributed.transaction.message.retry.ResendProcess;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareBatchMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.List;

/**
 * RabbitMQ抽象消息监听，所有消息消费者必须继承此类
 *
 * @author by Mr. Li 2020/12/12 0:22
 */
@Slf4j
public abstract class AbstractMessageListener implements ChannelAwareBatchMessageListener {
    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private ResendProcess resendProcess;

    /**
     * 所有子类 必须实现的业务类。
     *
     * @param message 消息对象
     */
    public abstract void receiveMessage(Message message);

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        // 记录总体消费次数
        Long consumerCount = getConsumerCount(message);

        log.info("消费者接受到消息，当前消息ID：{},消费次数：{}", message.getMessageProperties().getMessageId(),
                consumerCount);

        try {
            receiveMessage(message);

            // 成功的回执信息
            channel.basicAck(deliveryTag, false);

            // 消费成功，将Redis中统计信息消费次数的缓存删除
            redisTemplate.opsForHash().delete(MQConstants.MQ_CONSUMER_RETRY_COUNT_KEY, message.getMessageProperties().getMessageId());

        } catch (Exception e) {
            log.info("RabbitMQ 消费信息失败，{}", e.getMessage());
            // 判断是否超过最大重试次数
            if (consumerCount >= MQConstants.MAX_RETRY_COUNT ) {
                // 将消息扔进死信队列，稍后重试
                if (consumerCount <= MQConstants.MAX_CONSUMER_COUNT){ // 如果消费次数 大于 最大的消费次数。将走最后的 重试策略
                    resendProcess.resendReadyMsg();
                }
                channel.basicReject(deliveryTag, false);
            } else {
                //retry,按照2的指数级倍增
                Thread.sleep((long) (Math.pow(MQConstants.BASE_NUM, consumerCount) * 1000));
                getConsumerCount(message);
                // 重新回到队列
                channel.basicNack(deliveryTag, false, true);
            }
        }
    }



    private Long getConsumerCount(Message message) {
        Long consumerCount = redisTemplate.opsForHash().increment(MQConstants.MQ_CONSUMER_RETRY_COUNT_KEY,
                message.getMessageProperties().getMessageId(), 1);
        return consumerCount;
    }

    @Override
    public void onMessageBatch(List<Message> messages) {

    }

    @Override
    public void onMessageBatch(List<Message> messages, Channel channel) {

    }

    @Override
    public void onMessage(Message message) {

    }

    @Override
    public void containerAckMode(AcknowledgeMode mode) {

    }
}
